/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lvyh.lightframe.job.admin.executor;

import com.alibaba.dubbo.config.ApplicationConfig;
import com.alibaba.dubbo.config.ReferenceConfig;
import com.alibaba.dubbo.config.RegistryConfig;
import com.alibaba.dubbo.config.utils.ReferenceConfigCache;
import com.alibaba.dubbo.rpc.service.GenericService;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Maps;
import com.lvyh.lightframe.job.admin.common.RegistryContainer;
import com.lvyh.lightframe.job.admin.job.DubboJob;
import com.lvyh.lightframe.job.admin.util.NetUtils;
import com.lvyh.lightframe.job.admin.util.SpringContextUtils;
import com.lvyh.lightframe.job.client.InvokeContext;
import com.lvyh.lightframe.job.client.InvokeResult;
import org.apache.commons.lang3.StringUtils;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * Dubbo service scheduling executor
 */
@Service
public class DubboJobExecutor {

    private Logger logger = LoggerFactory.getLogger(DubboJobExecutor.class);

    @Resource
    private RegistryConfig registryConfig;

    public InvokeResult execute(JobExecutionContext context, DubboJob job) {
        if (context == null || job == null) {
            throw new RuntimeException("Dubbo task input parameter is null.");
        }
        InvokeResult invokeResult = new InvokeResult();
        try {
            JobDataMap dataMap = context.getMergedJobDataMap();
            if (dataMap == null) throw new RuntimeException("Task datamap is empty.");

            String executorRouteStrategy = (String) dataMap.get("executorRouteStrategy");
            String service = (String) dataMap.get("dubboService");
            String method = (String) dataMap.get("dubboMethod");
            String version = (String) dataMap.get("dubboVersion");
            String group = (String) dataMap.get("dubboGroup");
            Integer timeout = (Integer) dataMap.get("timeout");
            timeout = (timeout == null || timeout <= 0 || timeout > 120000) ? 120000 : timeout;
            String params = (String) dataMap.get("params");

            Map<String, String> paramMap = null;
            if (StringUtils.isBlank(params)) {
                paramMap = Maps.newHashMap();
            } else {
                paramMap = JSON.parseObject(params, Map.class);
            }

            Map<String, Object> invokeParams = Maps.newHashMap();
            invokeParams.put("source", NetUtils.getHostAddress());
            invokeParams.put("password", "");
            invokeParams.put("params", paramMap);

            GenericService genericService = null;

            //sharding broadcast
            if (Objects.equals(executorRouteStrategy, "sharding_broadcast")) {
                RegistryContainer registryContainer = SpringContextUtils.getBean(RegistryContainer.class);
                List<String> addressList = registryContainer.getProvidersByService(service);
                for (int i = 0; i < addressList.size(); i++) {
                    paramMap.put("sharding_index", String.valueOf(i));
                    paramMap.put("sharding_total", String.valueOf(addressList.size()));

                    String directUrl = "dubbo://" + addressList.get(i) + "/" + service;
                    genericService = getServiceReference(service, group, version, timeout, 0, directUrl);
                    if (genericService == null) {
                        logger.info("[DubboJobExecutor] dubbo service not found, service:{}", service);
                        continue;
                    }

                    Object rpcInvokeResult = genericService.$invoke(method, new String[]{InvokeContext.class.getName()},
                            new Object[]{invokeParams});

                    String json = JSON.toJSONString(rpcInvokeResult);
                    logger.info("[DubboJobExecutor] dubbo service execute success, rpcInvokeResult:{}", rpcInvokeResult);
                }

                invokeResult.setSuccess(true);
                String json = JSON.toJSONString(invokeResult);
                return JSON.parseObject(json, InvokeResult.class);
            }

            genericService = getServiceReference(service, group, version, timeout, 0, null);
            if (genericService == null) {
                invokeResult.setSuccess(false);
                invokeResult.setFailedCount(1);
                invokeResult.setSuccessCount(0);
                invokeResult.setCode("ServiceNotFound");
                invokeResult.setDescription("Dubbo service not found");
                return invokeResult;
            }

            Object rpcInvokeResult = genericService.$invoke(method, new String[]{InvokeContext.class.getName()},
                    new Object[]{invokeParams});

            String json = JSON.toJSONString(rpcInvokeResult);
            logger.info("[DubboJobExecutor] dubbo service execute success, rpcInvokeResult:{}", rpcInvokeResult);
            return JSON.parseObject(json, InvokeResult.class);

        } catch (Exception e) {
            logger.error("Dubbo-RPC-ERR：{}\r\nJobData={}", e.getMessage(), JSON.toJSONString(context.getJobDetail().getJobDataMap()));
            invokeResult.setSuccess(false);
            invokeResult.setFailedCount(1);
            invokeResult.setSuccessCount(0);
            invokeResult.setCode("RPC-ERR");
            invokeResult.setDescription(e.getMessage());
        }

        return invokeResult;
    }

    public GenericService getServiceReference(String service, String group, String version, int timeout, int retries, String directUrl) {

        ApplicationConfig application = new ApplicationConfig();
        application.setName("light-frame-job-admin");

        ReferenceConfig<GenericService> reference = null;
        reference = new ReferenceConfig<GenericService>();
        reference.setApplication(application);

        if (!StringUtils.isEmpty(directUrl)) {
            /**
             * sharding broadcast
             * Dubbo direct connection:reference.setUrl("dubbo://192.168.*.*:26987/com.xxx.OrderDataManager");
             */
            reference.setUrl(directUrl);
        } else {
            reference.setRegistry(registryConfig);
        }

        reference.setInterface(service);
        reference.setGeneric(true);
        reference.setTimeout(timeout);
        reference.setVersion(version);
        reference.setGroup(group);
        reference.setRetries(retries);
        reference.setCheck(false);

        ReferenceConfigCache cache = ReferenceConfigCache.getCache();
        GenericService genericService = cache.get(reference);
        return genericService;
    }
}
